package com.ferry.transport.netty.cluster;

import com.ferry.protocol.StatusCode;
import com.ferry.protocol.payload.RequestPayload;
import com.ferry.registry.domain.service.ServiceMetaData;
import com.ferry.transport.api.Directory;
import com.ferry.transport.api.channel.FutureListener;
import com.ferry.transport.api.channel.WrappedChannel;
import com.ferry.transport.api.channel.WrappedChannelGroup;
import com.ferry.transport.netty.consumer.Connector;
import com.ferry.transport.netty.consumer.Response;
import com.ferry.transport.netty.future.future.InvokeFuture;
import com.ferry.transport.netty.lb.LoadBalancer;
import com.ferry.transport.netty.provider.MessageWrapper;
import com.ferry.transport.netty.provider.Request;
import com.ferry.transport.netty.provider.RespWrapper;
import org.jetbrains.annotations.NotNull;

import java.util.concurrent.CopyOnWriteArrayList;


/**
 * Created by daiyong
 */
public abstract class AbstractClusterInvoker implements ClusterInvoker {
	private LoadBalancer loadBalancer;
	private Connector connector;

	public AbstractClusterInvoker(LoadBalancer loadBalancer, Connector connector) {
		this.loadBalancer = loadBalancer;
		this.connector = connector;
	}

	public WrappedChannelGroup select(ServiceMetaData serviceMetaData) {
		Directory directory = new Directory();
		directory.setGroup(serviceMetaData.getGroup());
		directory.setServiceProviderName(serviceMetaData.getServiceName());
		directory.setVersion(serviceMetaData.getVersion());

		CopyOnWriteArrayList groupList = connector.directory(directory);
		WrappedChannelGroup wrappedChannelGroup = loadBalancer.select(groupList);

		return wrappedChannelGroup;
	}

	protected InvokeFuture write(Request request) {
		InvokeFuture invokeFuture = new InvokeFuture();

		RequestPayload requestPayload = request.getRequestPayload();
		MessageWrapper messageWrapper = request.getMessageWrapper();
		WrappedChannelGroup wrappedChannelGroup = select(messageWrapper.getServiceMetaData());

		WrappedChannel wrappedChannel = wrappedChannelGroup.next();

		wrappedChannel.write(requestPayload, new FutureListener<WrappedChannel>() {

			@Override
			public void success(WrappedChannel channel) throws Exception {
				requestPayload.clear();
			}

			@Override
			public void failure(WrappedChannel channel, Throwable cause) throws Exception {
				requestPayload.clear();
				RespWrapper result = new RespWrapper();
				result.setError(cause);
				Response response = new Response(requestPayload.invokeId());
				response.getResponsePayload().status(StatusCode.CLIENT_ERROR.getCode());
				response.setRespWrapper(result);
			}
		});

		InvokeFuture.futures.put(requestPayload.invokeId(), invokeFuture);

		return invokeFuture;
	}

	public LoadBalancer getLoadBalancer() {
		return loadBalancer;
	}

	public void setLoadBalancer(LoadBalancer loadBalancer) {
		this.loadBalancer = loadBalancer;
	}

	public Connector getConnector() {
		return connector;
	}

	public void setConnector(Connector connector) {
		this.connector = connector;
	}
}
